package com.example.baselibrary.rx;

import io.reactivex.Flowable;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;

/**
 * 有背压处理（Backpressure）的 Rxbus by zqq on 2018/6/11.
 */

public class RxBus {

    private final FlowableProcessor<Object> rxBus;

    // PublishSubject只会把在订阅发生的时间点之后来自原始Flowable的数据发射给观察者
    private RxBus() {
        rxBus = PublishProcessor.create().toSerialized();
    }

    private static class Holder {
        private static final RxBus instance = new RxBus();
    }
    public static RxBus getInstance() {
        return Holder.instance;
    }

    // 提供了一个新的事件
    public void post(Object object) {
        rxBus.onNext(object);
    }

    // 根据传递的 tClass 类型返回特定类型(tClass)的 被观察者
    public <T> Flowable<T> toFlowable(Class<T> tClass) {
        return rxBus.ofType(tClass);
    }

    public Flowable<Object> toFlowable() {
        return rxBus;
    }

    public boolean hasSubscribers(){
        return rxBus.hasSubscribers();
    }

    //普通订阅解绑
//    public static void rxBusUnbind(CompositeDisposable disposable) {
//        if (null != disposable && !disposable.isDisposed()) {
//            disposable.clear();
//        }
//    }

}
